using Confluent.Kafka;
using Newtonsoft.Json;
using System;
using System.IO;
using System.Threading.Tasks;
using static Org.BouncyCastle.Math.EC.ECCurve;

namespace DataService.Helper.Kafka
{
    public class KafkaProducer
    {
        LogHelper logHelper = new LogHelper();
        /// <summary>
        /// 生产者
        /// </summary>
        public async void ProduceAsync(string server, string topic, string data)
        {
            var conf = new ConsumerConfig {
                BootstrapServers = server
            };

            using (var p = new ProducerBuilder<Null, string>(conf).Build())
            {
                var dr = await p.ProduceAsync(topic, new Message<Null, string> { Value = data });

                logHelper.Debug(data);
                logHelper.Info($"Delivered '{dr.Value}' to '{dr.TopicPartitionOffset}'");

            }
        }
        /// <summary>
        /// 生产者
        /// </summary>
        public void Produce(string server, string topic, string data)
        {
            try
            {
                var config = new ConsumerConfig {
                    BootstrapServers = server
                };
                using (var producer = new ProducerBuilder<Null, string>(config).Build())
                {
                    producer.Produce(topic, new Message<Null, string> { Value = data });
                    producer.Flush(TimeSpan.FromSeconds(1));
                }
            }
            catch (ProduceException<string, string> e)
            {
                Console.WriteLine(e);
            }
        }
        public async Task SendMessageAsync(string server, string topic, string data)
        {
            var config = new ConsumerConfig {
                BootstrapServers = server
            };
            using (var producer = new ProducerBuilder<Null, string>(config).Build())
            {
                var deliveryReport = await producer.ProduceAsync(topic, new Message<Null, string> { Value = data });

                logHelper.Info($"delivered to: {deliveryReport.TopicPartitionOffset}");
                logHelper.Info($"Kafka sended data:{topic}-{data}");
            }
        }
    }
}
